
package net.bwie.realtime.jtp.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 日志分流，使用侧边流实现，划分为错误日志、启动日志、浏览日志、曝光日志、动作日志
 * @author xuanyu
 * @date 2025/5/18
 */
public class AppLogSplitProcessFunction extends ProcessFunction<String, String> {

    /**
     * 侧输出流输出时标记OutputTag实例
     */
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;
    public AppLogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag,
                                      OutputTag<String> displayTag, OutputTag<String> actionTag){
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }
    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        // 第1步、解析JSON数据
        JSONObject jsonObject = JSON.parseObject(value);
        // 第2步、错误日志
        if(null != jsonObject.get("err")){

        }
        // 第3步、启动日志
        if(null != jsonObject.get("start")){

        }
        // 第4步、页面日志
        else{
            // 输出页面日志
            out.collect(jsonObject.toJSONString());
            // 第5步、曝光日志
            if(null != jsonObject.get("displays") && !jsonObject.getJSONObject("displays").isEmpty()){
            }
            // 第6步、动作日志
            if(null != jsonObject.get("actions") && !jsonObject.getJSONObject("actions").isEmpty()){
            }
        }
    }
}